package com.shujia.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.Source

object Demo21Broadcast {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("广播变量演示")
    val context = new SparkContext(conf)
    //====================================================
    //使用Scala的方式读取学生数据文件，将其转换以学号作为键的map集合，属于在Driver端的一个变量
    val studentsMap: Map[String, String] = Source.fromFile("spark/data/students.csv")
      .getLines()
      .toList
      .map((line: String) => {
        val infos: Array[String] = line.split(",")
        val stuInfo: String = infos.mkString(",")
        (infos(0), stuInfo)
      }).toMap

    val scoresRDD: RDD[String] = context.textFile("spark/data/score.txt")

    /**
     * 将studentsMap变成一个广播变量，让每一个将来需要执行关联的Executor中都有一份studentsMap数据
     * 避免了每次Task任务拉取都要附带一个副本，拉取的速度变快了，执行速度也就变快了
     *
     * 广播大变量
     */
    val studentsMapBroadcast: Broadcast[Map[String, String]] = context.broadcast(studentsMap)


    /**
     * 将Spark读取的分数RDD与外部变量学生Map集合进行关联
     * 循环遍历scoresRDD，将学号一样的学生信息关联起来
     */
//    val resMapRDD: RDD[(String, String)] = scoresRDD.map((score: String) => {
//      val id: String = score.split(",")(0)
//      //使用学号到学生map集合中获取学生信息
//      val studentInfo: String = studentsMap.getOrElse(id, "无学生信息")
//      (score, studentInfo)
//    })
//    resMapRDD.foreach(println)

    /**
     * 使用广播变量进行关联
     */
    val resMapRDD: RDD[(String, String)] = scoresRDD.map((score: String) => {
      val id: String = score.split(",")(0)
      val stuMap: Map[String, String] = studentsMapBroadcast.value
      //使用学号到学生map集合中获取学生信息
      val studentInfo: String = stuMap.getOrElse(id, "无学生信息")
      (score, studentInfo)
    })
    resMapRDD.foreach(println)



  }
}
